package org.apache.rocketmq.store;

import com.sun.jna.NativeLong;
import com.sun.jna.Pointer;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.*;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.*;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.QueueTypeUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.logfile.MappedFile;
import org.apache.rocketmq.store.util.LibC;
import org.rocksdb.RocksDBException;
import org.apache.rocketmq.store.MessageExtEncoder.PutMessageThreadLocal;
import sun.nio.ch.DirectBuffer;

import java.net.Inet6Address;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/25
 * @方法描述：存储生产者客户端发送过来的消息的核心类
 */
public class CommitLog implements Swappable {

    protected static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    //CommitLog文件文件结尾的魔术标识，读取到这个就意味着一个CommitLog文件到结尾了
    public final static int BLANK_MAGIC_CODE = -875286124;

    //CRC校验字段预留长度
    public static final int CRC32_RESERVED_LEN = MessageConst.PROPERTY_CRC32.length() + 1 + 10 + 1;

    //CommitLog文件的内存映射文件管理器
    protected final MappedFileQueue mappedFileQueue;

    //消息存储引擎
    protected final DefaultMessageStore defaultMessageStore;

    //消息刷新管理器，这个管理器就负责使用具体的刷新策略把消息刷新到硬盘上
    //所谓刷新策略其实就是同步刷新和异步刷新
    private final FlushManager flushManager;

    //冷数据检查器
    private final ColdDataCheckService coldDataCheckService;

    //用来把消息追加到映射文件中的回调方法，这个回调方法非常重要，可以说消息存储最核心的操作就是在这个回调方法中执行的
    private final AppendMessageCallback appendMessageCallback;

    //这是一个线程安全的线程私有变量，这个变量中封装了一个MessageExtEncoder对象
    //这个MessageExtEncoder对象就是用来把要存储的消息编码成二进制格式的
    //至于为什么把这个MessageExtEncoder对象存储到现成的私有变量中，在asyncPutMessage方法中我会为大家详细解释
    private final ThreadLocal<PutMessageThreadLocal> putMessageThreadLocal;

    //已经确认的消息偏移量，这个偏移量的概念比较模糊，在非集群模式下，这个偏移量其实就是映射文件中写入的消息的偏移量
    protected volatile long confirmOffset = -1L;

    private volatile long beginTimeInLock = 0;

    //消息存储锁，执行消息存储操作时要先获得锁
    protected final PutMessageLock putMessageLock;

    //主题队列锁
    protected final TopicQueueLock topicQueueLock;

    //执行同步刷新数据操作时要用到的硬盘状态监听器
    //其实就是判断数据同步刷新操作是否超时的
    private final FlushDiskWatcher flushDiskWatcher;

    //单个CommitLog文件大小，默认1GB
    protected int commitLogSize;

    //是否启用属性CRC校验，保障数据完整性，开启后会校验消息属性的CRC32，防止传输过程中数据被篡改
    private final boolean enabledAppendPropCRC;


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：构造方法
     */
    public CommitLog(final DefaultMessageStore messageStore) {

        //得到CommitLog文件的存储路径
        String storePath = messageStore.getMessageStoreConfig().getStorePathCommitLog();
        //创建mappedFileQueue对象，其实是这样的，CommitLog只是一个文件夹，文件夹里有多个存储了消息的文件
        //在内存映射技术下，每一个文件都会对应一个MappedFile文件映射对象，而mappedFileQueue对象则是用来管理这些MappedFile文件对象的
        this.mappedFileQueue = new MappedFileQueue(storePath,
                //得到每一个MappedFile文件的默认大小
                messageStore.getMessageStoreConfig().getMappedFileSizeCommitLog(),
                //使用内存映射文件分配组件创建mappedFile对象
                messageStore.getAllocateMappedFileService());
        this.defaultMessageStore = messageStore;
        this.flushManager = new DefaultFlushManager();
        this.coldDataCheckService = new ColdDataCheckService();
        this.appendMessageCallback = new DefaultAppendMessageCallback(defaultMessageStore.getMessageStoreConfig());
        putMessageThreadLocal = new ThreadLocal<PutMessageThreadLocal>() {
            @Override
            protected PutMessageThreadLocal initialValue() {
                //为每个线程创建独立的MessageExtEncoder消息编码器对象
                return new PutMessageThreadLocal(defaultMessageStore.getMessageStoreConfig());
            }
        };
        //创建消息存储锁
        this.putMessageLock = messageStore.getMessageStoreConfig()
                .isUseReentrantLockWhenPutMessage()
                //创建可重入锁
                ? new PutMessageReentrantLock()
                //创建自旋锁
                : new PutMessageSpinLock();
        this.flushDiskWatcher = new FlushDiskWatcher();
        this.topicQueueLock = new TopicQueueLock(messageStore.getMessageStoreConfig().getTopicQueueLockNum());
        //设置单个CommitLog文件大小
        this.commitLogSize = messageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
        //设置CRC校验配置标志
        this.enabledAppendPropCRC = messageStore.getMessageStoreConfig().isEnabledAppendPropCRC();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：加载CommitLog文件的方法，在这个方法中，会把mappedFileQueue对象加载了
     * 随着mappedFileQueue对象的加载，存储消息的本地文件就会被映射到内存中了
     */
    public boolean load() {
        //加载映射文件队列
        boolean result = this.mappedFileQueue.load();
        //执行文件队列自检操作，其实就是文件的连续性，判断每一个MappedFile文件是否是完整的
        this.mappedFileQueue.checkSelf();
        log.info("load commit log " + (result ? "OK" : "Failed"));
        return result;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：启动CommitLog的方法
     */
    public void start() {
        //启动消息刷盘管理器，这个管理器启动之后，就会定期把操作系统缓存页中的消息刷新到硬盘上
        this.flushManager.start();
        log.info("start commitLog successfully. storeRoot: {}", this.defaultMessageStore.getMessageStoreConfig().getStorePathRootDir());
        //启动硬盘状态监听器，设置守护线程
        flushDiskWatcher.setDaemon(true);
        flushDiskWatcher.start();
        //启动冷数据检查服务
        if (this.coldDataCheckService != null) {
            this.coldDataCheckService.start();
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把单条消息的最大字节大小更新到线程私有的编码器中
     */
    public void updateMaxMessageSize(PutMessageThreadLocal putMessageThreadLocal) {
        //从配置信息中获取最大消息大小
        int newMaxMessageSize = this.defaultMessageStore.getMessageStoreConfig().getMaxMessageSize();
        //如果新的最大消息大小大于等于10字节，且与当前编码器的最大消息体大小不同时，就把消息的最大字节大小更新到编码器中
        //这里其实是考虑到了在程序运行过程中，配置信息可能动态变更的情况，因为在程序运行的时候，可以通过控制台直接修改配置信息
        if (newMaxMessageSize >= 10 && putMessageThreadLocal.getEncoder().getMaxMessageBodySize() != newMaxMessageSize) {
            putMessageThreadLocal.getEncoder().updateEncoderBufferCapacity(newMaxMessageSize);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：该方法在第十三版本代码中先定义成这样，后面再逐步完善，因为CommitLog类的内容实在是太多了，一个版本代码根本迭代不完
     */
    public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {
        //计算消息体的CRC32校验码，用于后续消息完整性校验
        msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
        //定义一个消息追加结果变量，这个消息追加结果就会用来记录消息追加到内存映射文件的结果
        AppendMessageResult result = null;
        //获取存储信息统计服务，用于收集和更新消息存储相关的统计指标
        StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
        //得到消息的主题名称
        String topic = msg.getTopic();
        //默认设置消息版本为V1版本，消息版本决定了消息的序列化格式和存储结构
        msg.setVersion(MessageVersion.MESSAGE_VERSION_V1);
        //从配置信息中获取配置，判断是否要根据主题名称长度自动选择消息版本
        boolean autoMessageVersionOnTopicLen = this.defaultMessageStore.getMessageStoreConfig().isAutoMessageVersionOnTopicLen();
        //如果启用了上面的配置，那就判断一下当前要存储的消息所属主题的名称是否超过Byte.MAX_VALUE
        //如果超过了就把消息版本设置为V2版本
        if (autoMessageVersionOnTopicLen && topic.length() > Byte.MAX_VALUE) {
            msg.setVersion(MessageVersion.MESSAGE_VERSION_V2);
        }
        //得到消息产生的主机地址，检查是否为IPv6地址，如果是IPv6地址，设置对应的标志位
        InetSocketAddress bornSocketAddress = (InetSocketAddress) msg.getBornHost();
        if (bornSocketAddress.getAddress() instanceof Inet6Address) {
            msg.setBornHostV6Flag();
        }
        //得到存储消息的服务器地址，检查是否为IPv6地址
        InetSocketAddress storeSocketAddress = (InetSocketAddress) msg.getStoreHost();
        if (storeSocketAddress.getAddress() instanceof Inet6Address) {
            msg.setStoreHostAddressV6Flag();
        }
        //得到当前线程私有的PutMessageThreadLocal对象，这个PutMessageThreadLocal对象中持有了一个MessageExtEncoder消息编码对象
        //这个MessageExtEncoder对象就会把当前要存储的消息首先编码成二进制数据，然后写入到MessageExtEncoder的byteBuf缓冲区，这个byteBuf缓冲区中存储的消息就会被直接追加到内存映射文件中
        //分析到这里大家应该稍微清楚一些了，实际上要存储的消息在存储之前就会被MessageExtEncoder消息编码对象编码，并且存储在MessageExtEncoder消息编码对象内部的缓冲区中
        //那这里用到了线程私有对象，这是因为在并发情况下，如果有多个线程执行消息存储的操作，注意，程序执行到这里并没有获得同步锁呢。只有到真正存储消息的时候才会获得同步锁
        //这也就意味着程序执行到这里，多个线程都可以使用MessageExtEncoder消息编码对象对要存储的消息对象进行编码，然后把消息存储到MessageExtEncoder对象的缓冲区中
        //如果多个线程共用同一个MessageExtEncoder消息编码对象，而存储数据的缓冲区只有一个，那肯定就会出现并发问题，所以这里就为每一个线程都创建了一个私有的MessageExtEncoder消息编码对象
        //这个MessageExtEncoder对象就存储在线程的PutMessageThreadLocal私有对象中，现在大家应该清楚PutMessageThreadLocal对象的作用了
        PutMessageThreadLocal putMessageThreadLocal = this.putMessageThreadLocal.get();
        //在这里更新最大消息字节大小，防止编码后的消息超过消息最大大小
        updateMaxMessageSize(putMessageThreadLocal);
        //构建消息存储的唯一key，其实就是消息所属主题和消息要存储到的目标队列的Id，格式为主题名-队列Id
        String topicQueueKey = generateKey(putMessageThreadLocal.getKeyBuilder(), msg);
        //初始化在上锁期间执行操作耗费的时间为0，用于后续的性能统计
        long elapsedTimeInLock = 0;
        //初始化需要解锁的消息文件映射对象
        MappedFile unlockMappedFile = null;
        //从文件映射管理器中得到最新的映射文件对象，这时候大家也能理解了，其实MappedFile就是CommitLog存储消息的基础单元
        //CommitLog就是一个文件夹，该文件夹下有多个MappedFile文件存储消息，单个MappedFile文件存储消息的大小默认为1GB
        //消息写满了就可以创建新的MappedFile文件继续存储消息，所以这里要得到最新的MappedFile文件来存储消息
        MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
        //在这里获得主题队列的锁，使用的就是刚才构建的key，这就保证了在同一时刻，只能对同一主题下同一队列执行串行操作
        //确保了同一个主题下的同一个队列消息的顺序写入
        //这个是很重要的，因为后面消息重放的时候大家就清楚了，目标队列的Id其实就是消费队列的Id，消费队列的消息是按照顺序写入的，也就保证了后续消费消息的时候也是按照消息存储顺序消费的
        //如果要操作的是不同的目标队列，那么就会出现并发的情况，这一点大家要梳理清楚
        topicQueueLock.lock(topicQueueKey);
        try {
            //为消息分配消费队列偏移量，这里也可以看到，消息在消费队列的偏移量是在存储消息之前就已经定义好的了
            //并且这个顺序绝对是正确的，已经得到了同步锁，肯定能保证消息在同一主题同意队列中是顺序存储的
            defaultMessageStore.assignOffset(msg);
            //在这里从线程本地变量中得到消息编码器，对消息进行编码操作
            //这个时候消息就已经以二进制的格式存储在缓冲区中了，后面只要把缓冲区中的消息数据存放到映射文件的缓冲区即可
            PutMessageResult encodeResult = putMessageThreadLocal.getEncoder().encode(msg);
            //消息编码错误则返回错误结果
            if (encodeResult != null) {
                return CompletableFuture.completedFuture(encodeResult);
            }
            //把编码后的消息设置到消息对象中
            msg.setEncodedBuff(putMessageThreadLocal.getEncoder().getEncoderBuffer());
            //构建消息存储上下文信息对象
            PutMessageContext putMessageContext = new PutMessageContext(topicQueueKey);
            //在这里得到了消息存储同步锁，这个同步锁可以说是全局同步锁，因为接下来要把消息写入到CommitLog文件中了
            //队列这种逻辑概念在这里是不考虑的，为了保证消息存放到CommitLog文件的顺序性，这里必须得到全局锁，顺序存储消息
            putMessageLock.lock();
            try {
                //得到开始上锁的时间
                long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
                //把上锁时间赋值给beginTimeInLock成员变量
                this.beginTimeInLock = beginLockTimestamp;
                //设置消息存储时间，这个消息存储时间也反映了消息的存储顺序
                msg.setStoreTimestamp(beginLockTimestamp);
                //检查当前使用的MappedFile是否为null或已满，如果是这两种情况那就要创建新的内存映射文件了
                if (null == mappedFile || mappedFile.isFull()) {
                    //在这里创建新的内存映射文件，传入的这个参数0只是一个期望的文件起始偏移量
                    //等真正创建内存映射文件时，会根据真实的文件起始偏移量来创建
                    mappedFile = this.mappedFileQueue.getLastMappedFile(0);
                }
                //如果创建新文件失败
                if (null == mappedFile) {
                    //则记录错误日志
                    log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                    //重置上锁的起始时间为0，表示上锁后的锁操作已结束
                    beginTimeInLock = 0;
                    //返回消息存储失败结果，失败原因是内存映射文件创建失败
                    return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPPED_FILE_FAILED, null));
                }
                //执行到这里意味着内存映射文件创建成功了，那就可以使用消息追加回调对象把要存储的消息追加到文件映射对象中
                result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
                //根据消息追加的结果执行不同的操作
                switch (result.getStatus()) {
                    //如果是追加成功的情况
                    case PUT_OK:
                        //那就在这里继续执行消息存储的回调方法，该回调方法源码没有实现，可以由用户自己定义
                        onCommitLogAppend(msg, result, mappedFile);
                        break;
                    //如果追加结果是要追加的文件已经写满了
                    case END_OF_FILE:
                        //如果是这样的话，同样可以执行消息存储后的回调方法，只不过用户可以在该回调方法中根据追加结果
                        //执行不同的操作
                        onCommitLogAppend(msg, result, mappedFile);
                        //如果最新的内存映射文件已经写满了，那就记录一下这个文件
                        //如果开启了预热功能，后面就会释放该文件的内存锁，方便该文件对应的物理内存被释放
                        unlockMappedFile = mappedFile;
                        //这里就是创建新的内存映射文件的操作
                        mappedFile = this.mappedFileQueue.getLastMappedFile(0);
                        if (null == mappedFile) {
                            //内存映射文件创建失败则记录错误日志
                            log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                            beginTimeInLock = 0;
                            //返回创建内存映射文件失败的结果
                            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPPED_FILE_FAILED, result));
                        }
                        //在这里把消息追加到新的内存映射文件中
                        result = mappedFile.appendMessage(msg, this.appendMessageCallback, putMessageContext);
                        //如果在新文件中追加成功，则执行消息存储后的回调方法
                        if (AppendMessageStatus.PUT_OK.equals(result.getStatus())) {
                            onCommitLogAppend(msg, result, mappedFile);
                        }
                        break;
                    //这个是消息最大字节大小超过限制的情况
                    case MESSAGE_SIZE_EXCEEDED:
                        //这个是消息属性大小超过限制的情况
                    case PROPERTIES_SIZE_EXCEEDED:
                        beginTimeInLock = 0;
                        //返回要存储的消息非法的错误结果
                        return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));
                    //未知错误则进入下面分支
                    case UNKNOWN_ERROR:
                    default:
                        beginTimeInLock = 0;
                        return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
                }
                //计算在上锁期间执行操作耗费的时间，用于性能监控和统计
                elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
                //重置锁开始时间，表示上锁操作结束
                beginTimeInLock = 0;
            } finally {
                //释放消息存储锁
                putMessageLock.unlock();
            }
            //在这里再次判断一下消息是否成功追加到内存映射文件中
            if (AppendMessageStatus.PUT_OK.equals(result.getStatus())) {
                //如果成功追加到内存映射文件中，则增加对应消费队列的消息偏移量
                //也就是该队列存储了多少条消息的数量
                this.defaultMessageStore.increaseOffset(msg, getMessageNum(msg));
            }
            //如果使用的是RocksDB数据库进行消息存储，那么有异常的时候会在这里捕获异常
            //如果使用的是RocksDB数据库进行消息存储，那么有异常的时候会在这里捕获异常
        } catch (RocksDBException e) {
            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
        } finally {
            //释放主题队列锁
            topicQueueLock.unlock(topicQueueKey);
        }
        //判断上锁内的操作是否超过阈值
        if (elapsedTimeInLock > 500) {
            //超过阈值则记录告警日志，反应程序性能或者系统负载的情况
            log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}",
                    elapsedTimeInLock, msg.getBody().length, result);
        }
        //这里判断一下程序是否开启了映射文件预热功能，如果开启了就要释放对应的文件锁，注意，这里释放的是已经写满的内存映射文件的锁
        //至于内存映射文件预热，以及上锁和释放锁的相关内容，我会在对应的文章中讲解得非常详细，大家直接看文章即可
        if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
            //如果启用了预热映射文件功能，则在这里释放对应文件的锁
            this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
        }
        //创建最终的消息写入结果对象，结果状态设置为PUT_OK，表示消息已成功写入
        PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
        //更新存储统计服务中的主题消息计数器，记录特定主题的消息写入次数，用于监控和统计
        storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).add(result.getMsgNum());
        //更新存储统计服务中的主题消息大小计数器，记录特定主题的消息写入字节数，用于监控和统计
        storeStatsService.getSinglePutMessageTopicSizeTotal(topic).add(result.getWroteBytes());
        //到此为止消息应该已经成功追加到内存映射文件中了，那么接下来就应该处理后续得刷盘操作，以及集群模式下，消息同步给集群其他节点的操作
        //这些操作就是在下面这个方法中被执行的，当然，目前我们还没有引入集群，所以目前下面这个方法还不完善，只执行了和消息刷盘相关的操作
        return handleDiskFlushAndHA(putMessageResult, msg, 0, false);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到MessageExtBrokerInner对象中封装的消息的数量，该方法在批量存储消息的时候会发挥作用
     * 在第十四版本代码中，我们实现的存储单条消息的功能，所以MessageExtBrokerInner对象中只会有一条消息
     */
    protected short getMessageNum(MessageExtBrokerInner msgInner) {
        //初始化消息数量为1，也就是存储单挑消息的数量
        short messageNum = 1;
        //获取消息的消费队列类型
        CQType cqType = getCqType(msgInner);
        //判断当前要存储的消息是不是批量消息、内部批量消息，内部批量消息在我们的程序中并没有用到，源码中也没有用到
        if (MessageSysFlag.check(msgInner.getSysFlag(), MessageSysFlag.INNER_BATCH_FLAG) || CQType.BatchCQ.equals(cqType)) {
            if (msgInner.getProperty(MessageConst.PROPERTY_INNER_NUM) != null) {
                //如果是批量消息，那就在这里获取消息的实际数量
                messageNum = Short.parseShort(msgInner.getProperty(MessageConst.PROPERTY_INNER_NUM));
                //确保消息数量大于等于1
                messageNum = messageNum >= 1 ? messageNum : 1;
            }
        }
        return messageNum;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到消息的消费队列类型的方法
     */
    private CQType getCqType(MessageExtBrokerInner msgInner) {
        Optional<TopicConfig> topicConfig = this.defaultMessageStore.getTopicConfig(msgInner.getTopic());
        return QueueTypeUtils.getCQType(topicConfig);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消息追加到内存映射文件后要被回调的方法
     */
    protected void onCommitLogAppend(MessageExtBrokerInner msg, AppendMessageResult result, MappedFile commitLogFile) {
        this.getMessageStore().onCommitLogAppend(msg, result, commitLogFile);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：在消息成功写入到内存映射文件中后，就要执行对应的刷盘操作，如果是在集群模式下，并且当前节点是主节点，那还应该把消息同步给集群其他节点
     * 在当前方法中执行的就是这两个操作，HA其实就是高可用的意思
     */
    private CompletableFuture<PutMessageResult> handleDiskFlushAndHA(PutMessageResult putMessageResult,
                                                                     MessageExt messageExt, int needAckNums, boolean needHandleHA) {
        //在这里执行消息刷盘的操作，flushResultFuture封装了消息刷盘的结果
        //可以选择同步刷盘，也可以选择异步刷盘
        CompletableFuture<PutMessageStatus> flushResultFuture = handleDiskFlush(putMessageResult.getAppendMessageResult(), messageExt);
        //这里就是判断要不要执行同步消息的操作，如果不需要处理主从同步
        //声明一个CompletableFuture对象，用于保存主从同步操作的结果
        // CompletableFuture<PutMessageStatus> replicaResultFuture
        // if (!needHandleHA) {
        //     //那就在这里直接返回成功状态
        //     replicaResultFuture = CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
        // } else {
        //     //如果需要执行主从同步操作，那就在这里执行相关操作，并且返回执行结果
        //     replicaResultFuture = handleHA(putMessageResult.getAppendMessageResult(), putMessageResult, needAckNums);
        // }
        //

        //以上被注释掉的代码就是源码中的内容，在第十四版本代码中还没有引入集群，所以不必执行主从同步操作，所以我就先把上面的代码注释了
        //下面这行代码是我自己添加的，也就是没有执行主从同步操作的意思
        CompletableFuture<PutMessageStatus> replicaResultFuture =  CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
        //组合上面两个操作的执行结果
        return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> {
            //刷盘失败，更新结果状态
            if (flushStatus != PutMessageStatus.PUT_OK) {
                putMessageResult.setPutMessageStatus(flushStatus);
            }
            //主从同步失败，更新结果状态
            if (replicaStatus != PutMessageStatus.PUT_OK) {
                putMessageResult.setPutMessageStatus(replicaStatus);
            }
            return putMessageResult;
        });
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理消息刷盘的方法
     */
    private CompletableFuture<PutMessageStatus> handleDiskFlush(AppendMessageResult result, MessageExt messageExt) {
        //在这里调用消息刷盘管理器的方法执行消息刷盘操作
        return this.flushManager.handleDiskFlush(result, messageExt);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据消息在CommitLog文件中的偏移量和大小，得到消息存储时间的方法
     */
    public long pickupStoreTimestamp(final long offset, final int size) {
        //检查消息偏移量是否在合法范围之内，也就是判断一下消息偏移量是否超过了CommitLog文件中最小消息偏移量
        //是否小于CommitLog文件中最大的消息偏移量
        if (offset >= this.getMinOffset() && offset + size <= this.getMaxOffset()) {
            //如果在合法范围之内，那就根据指定消息的偏移量和消息大小从CommitLog文件中得到这条消息
            //SelectMappedBufferResult中的ByteBuffer就存储了消息内容
            SelectMappedBufferResult result = this.getMessage(offset, size);
            if (null != result) {
                try {//成功查找到指定的消息之后，先从ByteBuffer中读取消息的系统标志，因为这个系统标志中定义了生产消息的服务器的IP地址信息
                    //IP地址如果是IPv6，那么IP地址字节长度为20，这和IPv4字节长度是不一致的
                    int sysFlag = result.getByteBuffer().getInt(MessageDecoder.SYSFLAG_POSITION);
                    //这里判断消息的bornhost长度究竟是IPv4还是IPv6，如果是IPv4，那么 bornhostLength 为 8，否则为 20
                    int bornhostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 8 : 20;
                    //这里计算的就是消息存储时间在整个消息内容中的字节偏移量
                    int msgStoreTimePos = 4 + 4 + 4 + 4 + 4 + 8 + 8 + 4 + 8 + bornhostLength;
                    //根据具体的消息时间偏移量，从ByteBuffer中得到这条消息的存储时间
                    return result.getByteBuffer().getLong(msgStoreTimePos);
                } finally {
                    //释放SelectMappedBufferResult以免内存泄漏
                    result.release();
                }
            }
        }
        //偏移量不合法直接返回-1即可
        return -1;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据消息在CommitLog文件中的偏移量和大小，得到消息内容的方法
     */
    public SelectMappedBufferResult getMessage(final long offset, final int size) {
        //从配置信息中得到用户配置的映射文件的大小，比如说我们就定义一个映射文件大小为1G
        int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
        //先根据指定的消息偏移量从CommitLog文件夹下的所有内存映射文件中查找这条消息所在的内存映射文件
        //offset == 0则意味着如果找遍了所有内存映射文件，都没有找到对应的内存映射文件，那就返回第一个内存映射文件
        MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
        if (mappedFile != null) {
            //找到了对应的内存映射文件之后，根据消息在CommitLog中的全局偏移量，计算出这条消息在找到的内存映射文件中的偏移量
            //比如说消息的全局偏移量是2.5g字节，映射文件大小为1g字节，那么取模之后就可以得到该消息在映射文件中的偏移量为0.5g字节
            int pos = (int) (offset % mappedFileSize);
            //这里就是根据消息的大小和在内存映射文件中的具体偏移量，得到了存储该消息的缓冲区内容
            SelectMappedBufferResult selectMappedBufferResult = mappedFile.selectMappedBuffer(pos, size);
            if (null != selectMappedBufferResult) {
                //设置该消息是否在操作系统缓存页中的标志
                selectMappedBufferResult.setInCache(coldDataCheckService.isDataInPageCache(offset));
                //返回选择的缓冲区结果
                return selectMappedBufferResult;
            }
        }
        //如果未找到映射文件或未能选择到缓冲区内容，直接返回null
        return null;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据指定的偏移量计算下一个映射文件的起始偏移量的方法
     */
    public long rollNextFile(final long offset) {
        int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
        return offset + mappedFileSize - offset % mappedFileSize;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到CommitLog文件中存储消息的最小偏移量的方法，这里我要简单解释一下，CommitLog文件夹下有很多存储了消息的文件
     * 虽然存储第一条消息的时候，是从0偏移量开始存储的，但是随着程序的运行，过期的文件会被删除，所以最小的消息偏移量也是会随着时间变化的
     * CommitLog文件夹中第一个内存映射文件的第一条消息的偏移量就是最小偏移量
     */
    public long getMinOffset() {
        //得到内存映射文件中的第一个文件
        MappedFile mappedFile = this.mappedFileQueue.getFirstMappedFile();
        if (mappedFile != null) {
            //判断该文件是否还可用
            if (mappedFile.isAvailable()) {
                //如果可用那就返回该文件在CommitLog中的起始偏移量
                return mappedFile.getFileFromOffset();
            } else {
                //如果不可用，那就寻找下一个映射文件，返回下一个文件的起始偏移量
                return this.rollNextFile(mappedFile.getFileFromOffset());
            }
        }
        //没有找到内存映射文件则直接返回-1
        return -1;
    }


    //设置确认偏移量的方法，集群模式下，主从数据同步时会用到该方法
    public void setConfirmOffset(long phyOffset) {
        this.confirmOffset = phyOffset;
        //将新的确认偏移量写入存储检查点，系统重启时能够恢复到正确的确认位置
        this.defaultMessageStore.getStoreCheckpoint().setConfirmPhyOffset(confirmOffset);
    }


    //获取最后一个内存映射文件的起始偏移量方法
    public long getLastFileFromOffset() {
        MappedFile lastMappedFile = this.mappedFileQueue.getLastMappedFile();
        if (lastMappedFile != null) {
            if (lastMappedFile.isAvailable()) {
                return lastMappedFile.getFileFromOffset();
            }
        }
        return -1;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：生成消息主题队列唯一键的方法
     */
    public String generateKey(StringBuilder keyBuilder, MessageExt messageExt) {
        keyBuilder.setLength(0);
        keyBuilder.append(messageExt.getTopic());
        keyBuilder.append('-');
        keyBuilder.append(messageExt.getQueueId());
        return keyBuilder.toString();
    }

    //得到存储消息时开始上锁时间的方法
    public long getBeginTimeInLock() {
        return beginTimeInLock;
    }



    //获取单个CommitLog文件的大小的方法
    public int getCommitLogSize() {
        return commitLogSize;
    }

    //获取内存映射文件队列管理器的方法
    public MappedFileQueue getMappedFileQueue() {
        return mappedFileQueue;
    }

    //获取默认的消息存储引擎的方法
    public MessageStore getMessageStore() {
        return defaultMessageStore;
    }


    //释放内存映射文件对应内存的方法
    @Override
    public void swapMap(int reserveNum, long forceSwapIntervalMs, long normalSwapIntervalMs) {
        this.getMappedFileQueue().swapMap(reserveNum, forceSwapIntervalMs, normalSwapIntervalMs);
    }


    //判断内存映射文件队列是否为空的方法
    public boolean isMappedFilesEmpty() {
        return this.mappedFileQueue.isMappedFilesEmpty();
    }


    //刷新内存映射文件数据到本地硬盘的方法
    public long flush() {
        this.mappedFileQueue.commit(0);
        this.mappedFileQueue.flush(0);
        //返回已刷盘的位置
        return this.mappedFileQueue.getFlushedWhere();
    }


    //获取已刷盘消息的偏移量的方法
    public long getFlushedWhere() {
        return this.mappedFileQueue.getFlushedWhere();
    }


    //得到CommitLog文件中消息最大偏移量的方法
    public long getMaxOffset() {
        return this.mappedFileQueue.getMaxOffset();
    }
    /*
     *
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：计算尚未提交到内存映射缓冲区中的数据大小的方法
     */
    public long remainHowManyDataToCommit() {
        return this.mappedFileQueue.remainHowManyDataToCommit();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：计算已经提交给内存映射缓冲区但是还没有刷盘的数据大小的方法
     */
    public long remainHowManyDataToFlush() {
        return this.mappedFileQueue.remainHowManyDataToFlush();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：终止CommitLog组件工作的方法
     */
    public void shutdown() {
        //关闭消息刷盘管理器
        this.flushManager.shutdown();
        log.info("shutdown commitLog successfully. storeRoot: {}", this.defaultMessageStore.getMessageStoreConfig().getStorePathRootDir());
        //关闭同步刷盘监控器
        flushDiskWatcher.shutdown(true);
        //关闭冷数据检查器
        if (this.coldDataCheckService != null) {
            this.coldDataCheckService.shutdown();
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：销毁CommitLog组件的方法
     */
    public void destroy() {
        //调用映射文件队列的销毁方法，释放所有映射文件资源
        this.mappedFileQueue.destroy();
    }

    public void checkSelf() {
        mappedFileQueue.checkSelf();
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：默认的消息追加回调对象，就是在这个回调对象的回调方法中，会把要存储的消息追加到内存映射文件的缓冲区中
     */
    class DefaultAppendMessageCallback implements AppendMessageCallback {

        //内存映射文件写满的标志，或者说内存映射文件剩下的空间不足以再写入一条消息时，就会写入8个字节的文件结束标志，这8个字节就是预留出来的
        private static final int END_FILE_MIN_BLANK_LENGTH = 4 + 4;

        //堆外内存缓冲区，用来向内存映射文件写入文件结尾标志
        private final ByteBuffer msgStoreItemMemory;

        //CRC校验码预留长度
        private final int crc32ReservedLength = CommitLog.CRC32_RESERVED_LEN;

        private final MessageStoreConfig messageStoreConfig;

        DefaultAppendMessageCallback(MessageStoreConfig messageStoreConfig) {
            this.msgStoreItemMemory = ByteBuffer.allocate(END_FILE_MIN_BLANK_LENGTH);
            this.messageStoreConfig = messageStoreConfig;
        }


        //该方法并不会在我们的代码中用到，所以我就不添加注释了
        public AppendMessageResult handlePropertiesForLmqMsg(ByteBuffer preEncodeBuffer, final MessageExtBrokerInner msgInner) {
            if (msgInner.isEncodeCompleted()) {
                return null;
            }
            msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
            final byte[] propertiesData = msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);
            boolean needAppendLastPropertySeparator = enabledAppendPropCRC && propertiesData != null && propertiesData.length > 0 && propertiesData[propertiesData.length - 1] != MessageDecoder.PROPERTY_SEPARATOR;
            final int propertiesLength = (propertiesData == null ? 0 : propertiesData.length) + (needAppendLastPropertySeparator ? 1 : 0) + crc32ReservedLength;
            if (propertiesLength > Short.MAX_VALUE) {
                log.warn("putMessage message properties length too long. length={}", propertiesData.length);
                return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
            }
            int msgLenWithoutProperties = preEncodeBuffer.getInt(0);
            int msgLen = msgLenWithoutProperties + 2 + propertiesLength;
            if (msgLen > this.messageStoreConfig.getMaxMessageSize()) {
                log.warn("message size exceeded, msg total size: " + msgLen + ", maxMessageSize: " + this.messageStoreConfig.getMaxMessageSize());
                return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
            }
            preEncodeBuffer.putInt(0, msgLen);
            preEncodeBuffer.position(msgLenWithoutProperties);
            preEncodeBuffer.putShort((short) propertiesLength);
            if (propertiesLength > crc32ReservedLength) {
                preEncodeBuffer.put(propertiesData);
            }
            if (needAppendLastPropertySeparator) {
                preEncodeBuffer.put((byte) MessageDecoder.PROPERTY_SEPARATOR);
            }
            preEncodeBuffer.position(preEncodeBuffer.position() + crc32ReservedLength);
            msgInner.setEncodeCompleted(true);
            return null;
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：将要存储的消息追加到内存映射文件缓冲区中的方法，方法的第一个参数是内存映射文件的起始偏移量，第二个参数是内存映射文件使用的ByteBuffer
         * 第三个参数是内存映射文件剩余的最大可写空间，第四个参数是消息对象本身，第五个参数是消息存储上下文对象
         */
        public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
                                            final MessageExtBrokerInner msgInner, PutMessageContext putMessageContext) {

            //得到存储了编码后的消息的缓冲区
            ByteBuffer preEncodeBuffer = msgInner.getEncodedBuff();
            //得到消息的总长度，这里直接读取消息的前4个字节，这一点大家可以直接从MessageExtEncoder消息编码器的
            //encode()方法中得到印证，该方法中把消息编码的格式展示的非常清楚
            final int msgLen = preEncodeBuffer.getInt(0);
            //重置消息编码缓冲区，这就表示下一次再读取消息仍然是从头开始读取
            preEncodeBuffer.position(0);
            //设置消息读取的长度限制，这个限制就是消息的总长度
            preEncodeBuffer.limit(msgLen);
            //计算当前要存储的消息在CommitLog文件中的全局偏移量
            //fileFromOffset是存储消息的内存映射文件的起始偏移量，byteBuffer.position()是内存映射缓冲区要写入消息的具体位置
            long wroteOffset = fileFromOffset + byteBuffer.position();
            //构建消息Id生成器，这里的消息Id生成器会根据消息的存储主机地址和全局物理偏移量生成一个消息Id
            Supplier<String> msgIdSupplier = () -> {
                //判断是否使用IPv6地址格式
                int sysflag = msgInner.getSysFlag();
                //如果使用IPv6地址格式，那么消息Id的长度就是28个字节，如果是IPv4地址格式，那么消息Id的长度就是16个字节
                int msgIdLen = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 + 8 : 16 + 4 + 8;
                //分配存储消息Id的缓冲区
                ByteBuffer msgIdBuffer = ByteBuffer.allocate(msgIdLen);
                //得到存储消息的服务器地址，然后把服务器地址转换成二进制格式
                MessageExt.socketAddress2ByteBuffer(msgInner.getStoreHost(), msgIdBuffer);
                //重置msgIdBuffer缓冲区，这里clear方法并不会清除缓冲区的数据，只会重置写指针
                msgIdBuffer.clear();
                //在缓冲区最后8字节位置写入消息全局物理偏移量，这个时候就得到了新生成的消息Id
                msgIdBuffer.putLong(msgIdLen - 8, wroteOffset);
                //把消息Id转换为字符串格式
                return UtilAll.bytes2string(msgIdBuffer.array());
            };
            //得到消息在消费队列中的偏移量，这个偏移量之前已经计算完毕了
            Long queueOffset = msgInner.getQueueOffset();
            //得到消息的实际数量，在第十四版本代码中，我们实现的是存储单条消息功能
            //所以这里只会得到一条消息
            short messageNum = getMessageNum(msgInner);
            //以下这段和事务消息相关的操作先注释掉
            // final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
            // switch (tranType) {
            //     case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
            //     case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
            //         queueOffset = 0L;
            //         break;
            //     case MessageSysFlag.TRANSACTION_NOT_TYPE:
            //     case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
            //     default:
            //         break;
            // }

            //这里就要判断当前存储消息的内存映射文件剩余的最大可写容量是否够用
            //其实就是看看要存储的消息加上文件结束标志的长度是否超过了内存映射文件的最大可写容量
            if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
                //如果超过了就意味着这个内存映射文件不够用，这个时候就可以给这个内存映射文件直接写入结束标志
                //然后返回映射文件已满的结果，这样在外层方法就会创建新的内存映射文件存储消息了
                //清空msgStoreItemMemory缓冲区对象，重置其写指针
                this.msgStoreItemMemory.clear();
                //开始构建映射文件的结尾标志内容，把内存映射文件剩余的最大容量写入到缓冲区
                this.msgStoreItemMemory.putInt(maxBlank);
                //写入文件结尾魔术标志
                this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
                //因为内存映射文件满了，不能真的把消息存储到该文件中，所以要返回外层方法内存文件已满的结果
                //这里就象征性得到开始执行存储消息操作的时间
                final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
                //把文件已满的标志内容写入到内存文件的缓冲区中
                byteBuffer.put(this.msgStoreItemMemory.array(), 0, 8);
                //返回内存文件已满的结果，该构造方法中最后一个参数CommitLog.this.defaultMessageStore.now() - beginTimeMills就表示存储文件的耗时
                //性能监控会需要这些参数信息
                return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset,
                        maxBlank, /* 实际写入8字节但声明maxBlank长度用于位置计算 */
                        msgIdSupplier, msgInner.getStoreTimestamp(),
                        queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
            }

            //以下就是内存映射文件没满，真正把消息追加到内存映射文件缓冲区的操作
            //先定位到消息在消费队列偏移量的字节位置，这一点可以从MessageExtEncoder对象的encode()方法中印证
            //大家可以看看跳过的这四个字节分别表示什么内容
            int pos = 4 + 4 + 4 + 4 + 4;
            //把消息在消费队列中的偏移量写入到缓冲区中
            //这个消费队列偏移量又可以称为逻辑偏移量，因为消费队列本来就是逻辑上的概念
            //并不是真实存在的，CommitLog才是真实存在的，所以消息在CommitLog中的偏移量又被成为物理偏移量
            preEncodeBuffer.putLong(pos, queueOffset);
            //在这里累加写指针位置
            pos += 8;
            //接下来把消息的物理偏移量，也就是这条消息在CommitLog文件中的全局偏移量写入到缓冲区中
            preEncodeBuffer.putLong(pos, fileFromOffset + byteBuffer.position());
            //计算IP地址字节长度
            int ipLen = (msgInner.getSysFlag() & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
            //这里累加写指针位置，把消息全局偏移量、系统标志、消息产生的时间，消息存储服务器网络地址都跳过了
            pos += 8 + 4 + 8 + ipLen;
            //在这里把消息存储时间写入到缓冲区中
            preEncodeBuffer.putLong(pos, msgInner.getStoreTimestamp());
            //得到存储消息操作开始执行的时间
            final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
            //在这里把编码缓冲区中的内容直接添加到内存映射文件缓冲区中了
            byteBuffer.put(preEncodeBuffer);
            //清除编码缓冲区对象引用
            msgInner.setEncodedBuff(null);
            //在这里就可以给外层方法返回追加消息到缓冲区成功的结果了
            return new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgIdSupplier,
                    msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills, messageNum);
        }



        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：追加批量消息到内存映射文件缓冲区的方法，在第十四版本代码中同样用不到这个方法，所以我就先不给该方法添加注释了
         */
        public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank, final MessageExtBatch messageExtBatch, PutMessageContext putMessageContext) {
            byteBuffer.mark();
            long wroteOffset = fileFromOffset + byteBuffer.position();
            Long queueOffset = messageExtBatch.getQueueOffset();
            long beginQueueOffset = queueOffset;
            int totalMsgLen = 0;
            int msgNum = 0;
            final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
            ByteBuffer messagesByteBuff = messageExtBatch.getEncodedBuff();
            int sysFlag = messageExtBatch.getSysFlag();
            int bornHostLength = (sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
            int storeHostLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
            Supplier<String> msgIdSupplier = () -> {
                int msgIdLen = storeHostLength + 8;
                int batchCount = putMessageContext.getBatchSize();
                long[] phyPosArray = putMessageContext.getPhyPos();
                ByteBuffer msgIdBuffer = ByteBuffer.allocate(msgIdLen);
                MessageExt.socketAddress2ByteBuffer(messageExtBatch.getStoreHost(), msgIdBuffer);
                msgIdBuffer.clear();
                StringBuilder buffer = new StringBuilder(batchCount * msgIdLen * 2 + batchCount - 1);
                for (int i = 0; i < phyPosArray.length; i++) {
                    msgIdBuffer.putLong(msgIdLen - 8, phyPosArray[i]);
                    String msgId = UtilAll.bytes2string(msgIdBuffer.array());
                    if (i != 0) {
                        buffer.append(',');
                    }
                    buffer.append(msgId);
                }
                return buffer.toString();
            };
            messagesByteBuff.mark();
            int index = 0;
            while (messagesByteBuff.hasRemaining()) {
                final int msgPos = messagesByteBuff.position();
                final int msgLen = messagesByteBuff.getInt();
                totalMsgLen += msgLen;
                if ((totalMsgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
                    this.msgStoreItemMemory.clear();
                    this.msgStoreItemMemory.putInt(maxBlank);
                    this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
                    messagesByteBuff.reset();
                    byteBuffer.reset();
                    byteBuffer.put(this.msgStoreItemMemory.array(), 0, 8);
                    return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgIdSupplier, messageExtBatch.getStoreTimestamp(),
                            beginQueueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
                }
                int pos = msgPos + 20;
                messagesByteBuff.putLong(pos, queueOffset);
                pos += 8;
                messagesByteBuff.putLong(pos, wroteOffset + totalMsgLen - msgLen);
                pos += 8 + 4 + 8 + bornHostLength;
                messagesByteBuff.putLong(pos, messageExtBatch.getStoreTimestamp());
                if (enabledAppendPropCRC) {
                    int checkSize = msgLen - crc32ReservedLength;
                    ByteBuffer tmpBuffer = messagesByteBuff.duplicate();
                    tmpBuffer.position(msgPos).limit(msgPos + checkSize);
                    int crc32 = UtilAll.crc32(tmpBuffer);
                    messagesByteBuff.position(msgPos + checkSize);
                    MessageDecoder.createCrc32(messagesByteBuff, crc32);
                }
                putMessageContext.getPhyPos()[index++] = wroteOffset + totalMsgLen - msgLen;
                queueOffset++;
                msgNum++;
                messagesByteBuff.position(msgPos + msgLen);
            }
            messagesByteBuff.position(0);
            messagesByteBuff.limit(totalMsgLen);
            byteBuffer.put(messagesByteBuff);
            messageExtBatch.setEncodedBuff(null);
            AppendMessageResult result = new AppendMessageResult(
                    AppendMessageStatus.PUT_OK,
                    wroteOffset,
                    totalMsgLen,
                    msgIdSupplier,
                    messageExtBatch.getStoreTimestamp(),
                    beginQueueOffset,
                    CommitLog.this.defaultMessageStore.now() - beginTimeMills
            );
            result.setMsgNum(msgNum);
            return result;
        }

    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消息刷新器的抽象父类，注意，这个抽象父类继承了ServiceThread，这也就意味着该父类下的子类对象都是一个服务线程组件，一旦启动
     * 就要执行其内部的run方法
     */
    abstract class FlushCommitLogService extends ServiceThread {
        //服务器关闭时，要把为提交或未刷新的消息刷新到磁盘
        //这个常量定义的就是可尝试刷新或提交的最大次数
        protected static final int RETRY_TIMES_OVER = 10;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：用于定期提交消息的服务组件，注意，这里定期提交消息指的是把存放在TransientStorePool临时存储池中的消息提交到
     * 内存映射文件的缓冲区中，这个提交操作的具体作用大家一定要梳理清楚，不要以为是把消息提交到本地硬盘了。
     * 当然，如果程序本身没有启用临时存储池，那么就不会真的执行提交消息的操作。
     */
    class CommitRealTimeService extends FlushCommitLogService {

        //最新一次把消息提交给内存映射文件缓冲区的时间
        private long lastCommitTimestamp = 0;


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：得到服务名称的方法
         */
        @Override
        public String getServiceName() {
            if (CommitLog.this.defaultMessageStore.getBrokerConfig().isInBrokerContainer()) {
                return CommitLog.this.defaultMessageStore.getBrokerIdentity().getIdentifier() + CommitRealTimeService.class.getSimpleName();
            }
            return CommitRealTimeService.class.getSimpleName();
        }



        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：当前服务组件的核心方法，当前服务组件的核心方法，只要CommitRealTimeService组件一启动，这个run方法就会启动
         * 就会定期把存放在TransientStorePool临时存储池中的消息提交到内存映射文件的缓冲区中。如果程序没有启用临时存储池，那么当前组件就不会启动
         * 也就不会执行run方法，也就不会提交消息到映射文件缓冲区中了
         */
        @Override
        public void run() {
            //记录当前服务组件启动日志
            CommitLog.log.info(this.getServiceName() + " service started");
            //只要程序没有被终止，那就一直执行循环中的代码
            while (!this.isStopped()) {
                //就算是死循环，也不能让线程一直工作，假如没有要提交的数据，难道线程也要一遍遍空循环吗？
                //所以这里要从配置信息中得到一个时间间隔，每次循环之后，线程会等待这个时间之后，再执行第二次循环的操作
                int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog();
                //得到最少提交的页数，这里我给大家解释一下，内存映射是根据操作系统缓存页来映射的，内存映射文件映射的缓存页大小为4KB
                //当使用临时存储池的时候，其实就是把要存储的消息先存放在堆外缓冲区中，因为临时存储池的本质就是一个堆外缓冲区，当消息先被存储到临时存储池，也就是堆外缓冲区后
                //会定期被提交到内存映射文件的缓冲区中，也就是操作系统缓存页。而在执行这个操作之前就会先判断一下，要提交的数据是否达到最少提交页数要求，如果达到要求，那么就会执行物理提交
                //因为把数据最终提交到操作系统缓存页也涉及到了系统调用，会发生用户态和内核态的切换，所以这里会增加系统调用的开销，也会增加线程的调度开销，所以要在这里尽量做到批量提交的效果
                //也就要判断一下要提交的数据是否达到规定的页数了
                int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();
                //得到提交数据的最大时间间隔，这算是一个兜底操作，如果数据一直不满足最小页数要求，那么就会在固定时间间隔之后，强制执行提交数据操作
                int commitDataThoroughInterval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval();
                //得到提交操作开始执行的时间
                long begin = System.currentTimeMillis();
                //这里判断一下具体最后提交的时间间隔是否大于规定的提交时间间隔
                if (begin >= (this.lastCommitTimestamp + commitDataThoroughInterval)) {
                    this.lastCommitTimestamp = begin;
                    //如果大于了就把commitDataLeastPages参数设置为0，这里设置为0之后，在执行提交操作的时候不管页数够不够，都会强制提交了
                    commitDataLeastPages = 0;
                }
                try {
                    //在这里执行提交数据到内存映射文件的操作，commitDataLeastPages是规定提交的最小页数，如果数据没有超过这个页数，就不提交
                    //这里的返回值我要给大家解释一下，如果返回了false，就表示有数据被提交到内存映射文件的缓冲区中，如果返回了true，就表示可能没有数据被提交到内存映射文件的缓冲区中
                    //具体逻辑还要去mappedFileQueue的commit方法中查看
                    boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);
                    //得到提交操作结束的时间
                    long end = System.currentTimeMillis();
                    if (!result) {
                        //执行到这里意味着有数据被提交到内存映射文件的缓冲区中了
                        //在这里更新最后一次提交数据的时间
                        this.lastCommitTimestamp = end;
                        //有新的数据被提交到内存映射文件的缓冲区中了，这个时候要唤醒数据刷盘管理器，执行刷盘操作
                        CommitLog.this.flushManager.wakeUpFlush();
                    }
                    //如果本次提交操作超过500ms时记录警告日志
                    if (end - begin > 500) {
                        log.info("Commit data to file costs {} ms", end - begin);
                    }
                    //到此为止，循环就要结束了，在执行下一次循环之前，让线程休息一会
                    //等待固定时间之后再运行，这里大家应该也能意识到为什么上面要唤醒flushManager刷盘器了
                    //刷盘器也不是一直都工作，它也会定期休息。而这些让线程休息或者唤醒线程的方法，都是这些组件的父类ServiceThread提供的
                    this.waitForRunning(interval);
                } catch (Throwable e) {
                    CommitLog.log.error(this.getServiceName() + " service has exception. ", e);
                }
            }
            //执行到这里意味着跳出了刚才的线程，这也就意味着程序可能要关闭了，接下来要执行一些善后工作
            boolean result = false;
            //下面就是所谓的程序关闭重试机制，在这里循环执行数据提交操作，直到没有数据可提交或者超过最大重试次数
            //这里我要解释一下，只要是被提交到操作系统缓存页的数据，只要不是服务器宕机，电源没电这种问题
            //那么就算rocketmq程序退出，操作系统缓存页中的数据也不会丢失，而是会被正常刷新到本地硬盘
            //所以这里要在程序退出前尽可能多的把临时存储池中的数据提交到内存映射文件缓冲区中
            for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
                //在这里执行提交操作，如果返回值未true，则意味着已经没有新的数据可以被提交到内存映射文件的缓冲区中了
                //这个时候也可以退出循环了
                result = CommitLog.this.mappedFileQueue.commit(0);
                CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
            }
            //记录服务组件终止工作日志
            CommitLog.log.info(this.getServiceName() + " service end");
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：实时刷盘的服务组件，这个组件就负责把内存映射文件缓冲区的数据刷新到本地硬盘中，如果程序内部使用的是异步刷盘策略，那么就是这个组件负责数据刷盘
     */
    class FlushRealTimeService extends FlushCommitLogService {
        //最后一次刷盘时间
        private long lastFlushTimestamp = 0;
        //打印刷盘进度信息的计数器
        private long printTimes = 0;


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：实时刷盘服务组件的核心方法，当服务组件一启动，就会执行下面的方法，下面方法的逻辑和刚才分析的CommitRealTimeService组件的run方法逻辑类似，所以我就不添加
         * 特别详细的注释了，大家自己看看就行
         */
        @Override
        public void run() {
            //记录组件开始工作的日志信息
            CommitLog.log.info(this.getServiceName() + " service started");
            //同样在循环中执行刷盘操作，除非程序停止运行了
            while (!this.isStopped()) {
                //这里有一个是否开启定时刷盘操作的配置，如果是true，则表示开启定时刷盘操作
                //至于什么是定时刷盘操作，很快我就会为大家解释
                boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed();
                //得到刷盘时间间隔，也就是本次循环和下一次循环的时间间隔
                int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
                //得到最小的刷盘页数，这里的最小刷盘页数表示的是，只有当内存映射文件的缓冲区中的数据达到这个页数时，才会执行刷盘操作
                //这里的这个操作就更重要了，因为刷盘是IO操作，如果频繁对少量数据执行IO刷盘操作，很可能会影响程序性能
                int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();
                //得到刷盘的最大兜底时间，超过这个时间了，不管有没有达到足够页数，也要执行刷盘操作
                int flushPhysicQueueThoroughInterval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval();
                //是否打印刷盘进度信息的标志
                boolean printFlushProgress = false;
                //得到当前系统时间
                long currentTimeMillis = System.currentTimeMillis();
                //接下来就是对比距离上一次刷盘操作的时间间隔，如果超过了最大刷盘时间间隔，则强制要求立即刷盘
                if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {
                    //更新最后刷盘时间戳为当前时间
                    this.lastFlushTimestamp = currentTimeMillis;
                    //flushPhysicQueueLeastPages设置为0表示强制要求立即刷盘，忽略最小页数限制
                    flushPhysicQueueLeastPages = 0;
                    //这压力有一点不一样的，那就是每经过十次最大刷盘时间间，就会打印一次刷盘进度日志
                    printFlushProgress = (printTimes++ % 10) == 0;
                }
                //下面就是刷盘的具体操作了
                try {
                    //这里就能体现出刷盘操作定时和不定时的区别了
                    if (flushCommitLogTimed) {
                        //这里就是定时刷盘的操作，让线程进入休眠，休眠时间是固定的刷盘时间间隔
                        Thread.sleep(interval);
                    } else {
                        //这里也是让线程进入休眠，休眠时间是固定的刷盘时间间隔，但是这个休眠可以随时被外部线程唤醒
                        //也就是说，虽然这里休眠了，但不一定真的会休眠这么久，只要有数据被写入到内存映射文件缓冲区中了，可能就会被其他线程唤醒，执行一次刷盘操作
                        //当然，具体是否刷盘，还要看待刷盘的数据是否超过了最小缓存页限制
                        //当然，这里可能会有些朋友感到困惑，为什么还没开始刷盘呢，就让线程休眠了
                        //这是因为程序刚启动的时候，也许内存映射文件中根本没有待刷盘的数据，所以倒不如先让线程休眠
                        //反正线程休眠了，只要有数据来了，该线程还会被唤醒
                        this.waitForRunning(interval);
                    }
                    //打印刷盘进度
                    if (printFlushProgress) {
                        this.printFlushProgress();
                    }

                    //记录刷盘操作开始时间
                    long begin = System.currentTimeMillis();
                    //执行物理刷盘的操作
                    CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages);
                    //得到数据最新的刷盘时间
                    long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
                    if (storeTimestamp > 0) {
                        //把消息最新的刷盘时间设置到存储检查点中，用于程序重启或故障恢复
                        CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
                    }
                    //得到刷盘操作的总耗时
                    long past = System.currentTimeMillis() - begin;
                    //耗时过长记录日志信息
                    if (past > 500) {
                        log.info("Flush data to disk costs {} ms", past);
                    }
                } catch (Throwable e) {
                    CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
                    this.printFlushProgress();
                }
            }
            //接下来就是程序关闭之前的操作了
            boolean result = false;
            //要重试几次，尽可能对的把数据刷盘
            for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
                //强制刷盘，如果返回值为true，则意味着所有数据已经刷盘完毕了
                result = CommitLog.this.mappedFileQueue.flush(0);
                CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
            }
            //打印刷盘信息
            this.printFlushProgress();
            CommitLog.log.info(this.getServiceName() + " service end");
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：得到服务名称的方法
         */
        @Override
        public String getServiceName() {
            if (CommitLog.this.defaultMessageStore.getBrokerConfig().isInBrokerContainer()) {
                return CommitLog.this.defaultMessageStore.getBrokerConfig().getIdentifier() + FlushRealTimeService.class.getSimpleName();
            }
            return FlushRealTimeService.class.getSimpleName();
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：打印刷盘进度信息的方法
         */
        private void printFlushProgress() {
            //该方法在源码中就被注释了，是一个预留的调试接口
            //自己调试程序的时候，可以把下面的代码取消注释
            // CommitLog.log.info("how much disk fall behind memory, "
            // + CommitLog.this.mappedFileQueue.howMuchFallBehind());
        }

        //等待线程结束工作的最大时间
        @Override
        public long getJoinTime() {
            return 1000 * 60 * 5;
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：同步刷盘组件，如果程序内部启用的是同步刷盘策略，那就是这个组件负责刷盘操作
     */
    class GroupCommitService extends FlushCommitLogService {


        //写请求队列，当执行同步刷盘操作的时候，会把刷盘操作封装成GroupCommitRequest请求，提交到这个队列中
        private volatile LinkedList<GroupCommitRequest> requestsWrite = new LinkedList<>();

        //当处理同步刷盘的GroupCommitRequest请求时，会把requestsWrite队列中的请求移动到requestsRead队列中中
        //这样一来，requestsWrite队列就空了，那么在处理同步刷盘操作时，requestsWrite队列还可以继续接收新的刷盘请求
        private volatile LinkedList<GroupCommitRequest> requestsRead = new LinkedList<>();

        private final PutMessageSpinLock lock = new PutMessageSpinLock();

        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：添加同步刷盘请求到requestsWrite队列中的方法
         */
        public void putRequest(final GroupCommitRequest request) {
            lock.lock();
            try {
                this.requestsWrite.add(request);
            } finally {
                lock.unlock();
            }
            this.wakeup();
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：交换读写队列内容的方法
         */
        private void swapRequests() {
            lock.lock();
            try {
                LinkedList<GroupCommitRequest> tmp = this.requestsWrite;
                this.requestsWrite = this.requestsRead;
                this.requestsRead = tmp;
            } finally {
                lock.unlock();
            }
        }

        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：同步刷新消息到本地硬盘的操作
         */
        private void doCommit() {
            //判断读队列是否有待处理的提交请求
            if (!this.requestsRead.isEmpty()) {
                //如果有请求，说明内存映射文件的缓冲区中存在一些数据要刷新到本地硬盘中
                for (GroupCommitRequest req : this.requestsRead) {
                    //这里做了一个比较，首先我们要清楚req.getNextOffset()代表的是什么，如果大家已经读完对应的文章了，肯定就都清楚了
                    //当要存储的消息被追加到内存映射文件缓冲区时，会返回一个AppendMessageResult结果对象，该结果的WroteOffset和WroteBytes成员变量分别表示这条消息在CommitLog文件中的全局偏移量和消息本身的长度
                    //而req.getNextOffset()其实就时AppendMessageResult对象的WroteOffset和WroteBytes的和，这个和就表示当前要刷盘的消息在刷盘之后的CommitLog文件的最大偏移量
                    //这里判断了一下，如果当前CommitLog文件最大偏移量大于req.getNextOffset()，则说明该请求对应的消息已经持久化到本地硬盘中，所以就不用执行刷盘操作了
                    //如果返回false，那就意味着还没有刷新本条消息，那就可以执行刷盘操作了
                    boolean flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
                    //这里有一个循环，当flushOK为false时，就会执行刷盘操作，循环次数是刷盘最大重试次数，应对磁盘高负载情景
                    //确保本次刷盘可以成功
                    for (int i = 0; i < 1000 && !flushOK; i++) {
                        //执行刷盘操作
                        CommitLog.this.mappedFileQueue.flush(0);
                        //判断是否还需要继续刷盘，返回true就意味着不需要再刷盘了
                        flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
                        if (flushOK) {
                            //在这里跳出循环
                            break;
                        } else {
                            try {
                                //如果返回false，则在这里睡眠1ms再刷盘，源码中的注释写得很清楚：当启用transientStorePool，也就是堆外内存缓存池时，消息可能尚未提交到页缓存
                                //通过短暂休眠1ms等待操作系统完成页缓存同步，避免线程空循环
                                // When transientStorePoolEnable is true, the messages in writeBuffer may not be committed
                                // to pageCache very quickly, and flushOk here may almost be false, so we can sleep 1ms to
                                // wait for the messages to be committed to pageCache.
                                Thread.sleep(1);
                            } catch (InterruptedException ignored) {

                            }
                        }
                    }
                    //以上操作执行完毕后，就意味着本次同步刷盘的操作成功了，当然，也可能意味着操作超时了这个时候就可以唤醒等待同步操作完成的县城了
                    //注意，GroupCommitRequest的wakeupCustomer方法就会给内部的CompletableFuture成员变量设置完成结果
                    //那么调用了get方法的外层方法就可以结束阻塞，继续向下执行了
                    req.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT);
                }
                //得到消息最新存储的时间戳
                long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
                if (storeTimestamp > 0) {
                    //更新存储检查点的时间
                    CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
                }
                //清空读队列
                this.requestsRead = new LinkedList<>();
            } else {
                //执行到这里意味着读队列是空的，那就在这里执行一次基础刷盘操作
                CommitLog.this.mappedFileQueue.flush(0);
            }
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：同步刷盘组件启动之后，就会执行该方法
         */
        @Override
        public void run() {
            CommitLog.log.info(this.getServiceName() + " service started");
            //在循环中执行刷新数据到硬盘的操作
            while (!this.isStopped()) {
                try {
                    //在这里等待结束的时候，会执行onWaitEnd()方法，这一点可以从父类中查看
                    //而在onWaitEnd()方法中，又会执行子类的swapRequests()方法，把写队列的请求交换到读队列中
                    //接下来就可以执行刷盘数据的操作了
                    //参数10意味着最多等待10ms
                    this.waitForRunning(10);
                    //执行真正刷新的操作
                    this.doCommit();
                } catch (Exception e) {
                    CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }
            try {
                //程序要停止运行的时候在这里等待一会，尽可能多的接收一些停机前的同步刷盘请求
                Thread.sleep(10);
            } catch (InterruptedException e) {
                CommitLog.log.warn("GroupCommitService Exception, ", e);
            }
            //在停机之前最后一次执行队列交换操作，然后执行刷盘操作
            this.swapRequests();
            //执行最后一次刷盘操作确保数据持久化
            this.doCommit();
            CommitLog.log.info(this.getServiceName() + " service end");
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：当前方法就是父类的waitForRunning方法执行结束之前的回调方法
         */
        @Override
        protected void onWaitEnd() {
            //执行队列交换操作，将写队列数据交给读队列使用
            this.swapRequests();
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：得到服务名称的方法
         */
        @Override
        public String getServiceName() {
            if (CommitLog.this.defaultMessageStore.getBrokerConfig().isInBrokerContainer()) {
                return CommitLog.this.defaultMessageStore.getBrokerConfig().getIdentifier() + GroupCommitService.class.getSimpleName();
            }
            return GroupCommitService.class.getSimpleName();
        }

        //等待线程结束工作的最大时间
        @Override
        public long getJoinTime() {
            return 1000 * 60 * 5;
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：默认的消息刷盘管理器
     */
    class DefaultFlushManager implements FlushManager {

        //异步刷盘服务组件
        private final FlushCommitLogService flushCommitLogService;

        //同步刷盘服务组件
        private final FlushCommitLogService commitRealTimeService;

        //构造方法
        public DefaultFlushManager() {
            //判断用户配置的刷盘方式是同步还是异步
            if (FlushDiskType.SYNC_FLUSH == CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
                //如果是同步刷盘方式，那就创建GroupCommitService同步刷盘对象
                this.flushCommitLogService = new CommitLog.GroupCommitService();
            } else {
                //如果是异步刷盘，则创建FlushRealTimeService异步刷盘服务组件
                this.flushCommitLogService = new CommitLog.FlushRealTimeService();
            }
            //创建堆外内存消息实时提交到映射文件缓冲区的服务组件
            this.commitRealTimeService = new CommitLog.CommitRealTimeService();
        }

        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：启动刷盘管理器的方法
         */
        @Override
        public void start() {
            //启动刷盘服务组件
            this.flushCommitLogService.start();
            //判断程序是否启用了临时存储池功能，如果启用了才启动commitRealTimeService组件工作
            if (defaultMessageStore.isTransientStorePoolEnable()) {
                this.commitRealTimeService.start();
            }
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：处理硬盘刷新操作的方法
         */
        public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {

            //判断是不是同步刷盘操作
            if (FlushDiskType.SYNC_FLUSH == CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
                //如果是同步刷盘操作，那就把flushCommitLogService转换成同步刷盘组件
                final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
                //检查当前要刷盘的消息是否要等待存储确认，同步刷盘肯定就需要等待了
                if (messageExt.isWaitStoreMsgOK()) {
                    //所以这里会创建一个同步刷盘请求
                    GroupCommitRequest request = new GroupCommitRequest(
                            result.getWroteOffset() + result.getWroteBytes(),
                            CommitLog.this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout()
                    );
                    //把请求提交到刷盘对象的写请求队列中
                    service.putRequest(request);
                    //得到封装刷盘结果的CompletableFuture对象
                    CompletableFuture<PutMessageStatus> flushOkFuture = request.future();
                    PutMessageStatus flushStatus = null;
                    try {
                        //在这里同步等待刷盘结果
                        flushStatus = flushOkFuture.get(
                                CommitLog.this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout(),
                                TimeUnit.MILLISECONDS
                        );
                    } catch (InterruptedException | ExecutionException | TimeoutException e) {

                    }
                    //执行到这里意味着阻塞结束，那就要检查刷盘操作是否成功了
                    if (flushStatus != PutMessageStatus.PUT_OK) {
                        //如果没有成功则记录错误日志
                        log.error("do groupcommit, wait for flush failed, topic: " + messageExt.getTopic()
                                + " tags: " + messageExt.getTags()
                                + " client address: " + messageExt.getBornHostString());
                        //设置同步刷盘超时状态
                        putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
                    }
                } else {
                    //如果不需要等待确认，那就直接唤醒同步刷盘组件，让该组件执行刷盘操作
                    service.wakeup();
                }
            }
            //执行到这里就是异步刷盘的逻辑，异步刷盘显然就不要要等待刷盘结果了
            else {
                //在这里先判断一下是否启用了临时存储池
                if (!CommitLog.this.defaultMessageStore.isTransientStorePoolEnable()) {
                    //没有启用则直接唤醒异步刷盘组件执行异步刷盘操作
                    flushCommitLogService.wakeup();
                } else {
                    //如果启用了临时存储池功能，则唤醒实时提交服务，这样就可以把要刷盘的消息从堆外内存提交到内存映射文件的缓冲区
                    //然后被异步刷盘组件刷新到本地硬盘了
                    commitRealTimeService.wakeup();
                }
            }
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：处理硬盘刷新操作的方法，该方法和上面方法的逻辑几乎一致，大家简单看看就行
         */
        @Override
        public CompletableFuture<PutMessageStatus> handleDiskFlush(AppendMessageResult result, MessageExt messageExt) {
            //判断是不是同步刷盘操作
            if (FlushDiskType.SYNC_FLUSH == CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
                //如果是同步刷盘操作，那就把flushCommitLogService转换成同步刷盘组件
                final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
                //检查当前要刷盘的消息是否要等待存储确认，同步刷盘肯定就需要等待了
                if (messageExt.isWaitStoreMsgOK()) {
                    //创建一个同步刷盘请求
                    GroupCommitRequest request = new GroupCommitRequest(
                            result.getWroteOffset() + result.getWroteBytes(),
                            CommitLog.this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout()
                    );
                    //这里还有一个操作需要解释一下，那就是把同步刷盘请求提交给硬盘监控器
                    //这样一来硬盘监控器就可以监听到该请求，发现同步刷盘操作超时之后可以立即唤醒等待同步操作完成的线程
                    flushDiskWatcher.add(request);
                    //把请求提交到刷盘对象的写请求队列中
                    service.putRequest(request);
                    return request.future();
                } else {
                    //如果不需要等待确认，那就直接唤醒同步刷盘组件，让该组件执行刷盘操作
                    service.wakeup();
                    //立即返回成功状态
                    return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
                }
            }
            //以下就是异步刷盘操作
            else {
                //在这里先判断一下是否启用了临时存储池
                if (!CommitLog.this.defaultMessageStore.isTransientStorePoolEnable()) {
                    //没有启用则直接唤醒异步刷盘组件执行异步刷盘操作
                    flushCommitLogService.wakeup();
                } else {
                    //如果启用了临时存储池功能，则唤醒实时提交服务，这样就可以把要刷盘的消息从堆外内存提交到内存映射文件的缓冲区
                    //然后被异步刷盘组件刷新到本地硬盘了
                    commitRealTimeService.wakeup();
                }
                //异步刷盘默认返回成功状态
                return CompletableFuture.completedFuture(PutMessageStatus.PUT_OK);
            }
        }

        //唤醒刷盘服务组件的方法
        @Override
        public void wakeUpFlush() {
            flushCommitLogService.wakeup();
        }

        //唤醒实时提交服务组件的方法
        @Override
        public void wakeUpCommit() {
            commitRealTimeService.wakeup();
        }

        //关闭刷盘管理器的方法
        @Override
        public void shutdown() {
            if (defaultMessageStore.isTransientStorePoolEnable()) {
                this.commitRealTimeService.shutdown();
            }
            this.flushCommitLogService.shutdown();
        }

    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：这个冷热数据检查器我就先不添加注释了，现在要讲解地内容已经够多了，如果我在文章中确定了要讲解这个检查器，那么到时候我会把对应的代码注释完善了
     * 大家可以先忽略这个服务组件
     */
    public class ColdDataCheckService extends ServiceThread {

        private final SystemClock systemClock = new SystemClock();

        private final ConcurrentHashMap<String, byte[]> pageCacheMap = new ConcurrentHashMap<>();

        private int pageSize = -1;

        private int sampleSteps = 32;

        public ColdDataCheckService() {
            sampleSteps = defaultMessageStore.getMessageStoreConfig().getSampleSteps();
            if (sampleSteps <= 0) {
                sampleSteps = 32;
            }
            initPageSize();
            scanFilesInPageCache();
        }


        @Override
        public String getServiceName() {
            return ColdDataCheckService.class.getSimpleName();
        }

        @Override
        public void run() {
            log.info("{} service started", this.getServiceName());
            while (!this.isStopped()) {
                try {
                    if (MixAll.isWindows()
                            || !defaultMessageStore.getMessageStoreConfig().isColdDataFlowControlEnable()
                            || !defaultMessageStore.getMessageStoreConfig().isColdDataScanEnable()) {
                        pageCacheMap.clear();
                        this.waitForRunning(180 * 1000);
                        continue;
                    } else {
                        this.waitForRunning(defaultMessageStore.getMessageStoreConfig().getTimerColdDataCheckIntervalMs());
                    }
                    if (pageSize < 0) {
                        initPageSize();
                    }
                    long beginClockTimestamp = this.systemClock.now();
                    scanFilesInPageCache();
                    long costTime = this.systemClock.now() - beginClockTimestamp;
                    log.info("[{}] scanFilesInPageCache-cost {} ms.",
                            costTime > 30 * 1000 ? "NOTIFYME" : "OK", costTime);
                } catch (Throwable e) {
                    log.warn(this.getServiceName() + " service has e: {}", e);
                }
            }
            log.info("{} service end", this.getServiceName());
        }


        public boolean isDataInPageCache(final long offset) {
            if (!defaultMessageStore.getMessageStoreConfig().isColdDataFlowControlEnable()) {
                return true;
            }
            if (pageSize <= 0 || sampleSteps <= 0) {
                return true;
            }
            if (!defaultMessageStore.checkInColdAreaByCommitOffset(offset, getMaxOffset())) {
                return true;
            }
            if (!defaultMessageStore.getMessageStoreConfig().isColdDataScanEnable()) {
                return false;
            }
            MappedFile mappedFile = mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
            if (null == mappedFile) {
                return true;
            }
            byte[] bytes = pageCacheMap.get(mappedFile.getFileName());
            if (null == bytes) {
                return true;
            }
            int pos = (int) (offset % defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog());
            int realIndex = pos / pageSize / sampleSteps;
            return bytes.length - 1 >= realIndex && bytes[realIndex] != 0;
        }


        private void scanFilesInPageCache() {
            if (MixAll.isWindows()
                    || !defaultMessageStore.getMessageStoreConfig().isColdDataFlowControlEnable()
                    || !defaultMessageStore.getMessageStoreConfig().isColdDataScanEnable()
                    || pageSize <= 0) {
                return;
            }
            try {
                log.info("pageCacheMap key size: {}", pageCacheMap.size());
                clearExpireMappedFile();
                mappedFileQueue.getMappedFiles().forEach(mappedFile -> {
                    byte[] pageCacheTable = checkFileInPageCache(mappedFile);
                    if (sampleSteps > 1) {
                        pageCacheTable = sampling(pageCacheTable, sampleSteps);
                    }
                    pageCacheMap.put(mappedFile.getFileName(), pageCacheTable);
                });
            } catch (Exception e) {
                log.error("scanFilesInPageCache exception", e);
            }
        }



        private void clearExpireMappedFile() {
            Set<String> currentFileSet = mappedFileQueue.getMappedFiles().stream()
                    .map(MappedFile::getFileName)
                    .collect(Collectors.toSet());

            pageCacheMap.forEach((key, value) -> {
                if (!currentFileSet.contains(key)) {
                    pageCacheMap.remove(key);
                    log.info("clearExpireMappedFile fileName: {}, has been clear", key);
                }
            });
        }


        private byte[] sampling(byte[] pageCacheTable, int sampleStep) {
            byte[] sample = new byte[(pageCacheTable.length + sampleStep - 1) / sampleStep];
            for (int i = 0, j = 0; i < pageCacheTable.length && j < sample.length; i += sampleStep) {
                sample[j++] = pageCacheTable[i];
            }
            return sample;
        }


        private byte[] checkFileInPageCache(MappedFile mappedFile) {
            long fileSize = mappedFile.getFileSize();
            final long address = ((DirectBuffer) mappedFile.getMappedByteBuffer()).address();
            int pageNums = (int) (fileSize + this.pageSize - 1) / this.pageSize;
            byte[] pageCacheRst = new byte[pageNums];
            int mincore = LibC.INSTANCE.mincore(new Pointer(address), new NativeLong(fileSize), pageCacheRst);
            if (mincore != 0) {
                log.error("checkFileInPageCache call the LibC.INSTANCE.mincore error, fileName: {}, fileSize: {}",
                        mappedFile.getFileName(), fileSize);
                for (int i = 0; i < pageNums; i++) {
                    pageCacheRst[i] = 1;
                }
            }
            return pageCacheRst;
        }


        private void initPageSize() {
            if (pageSize < 0 && defaultMessageStore.getMessageStoreConfig().isColdDataFlowControlEnable()) {
                try {
                    if (!MixAll.isWindows()) {
                        pageSize = LibC.INSTANCE.getpagesize();
                    } else {
                        defaultMessageStore.getMessageStoreConfig().setColdDataFlowControlEnable(false);
                        log.info("windows os, coldDataCheckEnable force setting to be false");
                    }
                    log.info("initPageSize pageSize: {}", pageSize);
                } catch (Exception e) {
                    defaultMessageStore.getMessageStoreConfig().setColdDataFlowControlEnable(false);
                    log.error("initPageSize error, coldDataCheckEnable force setting to be false ", e);
                }
            }
        }


        // public boolean isMsgInColdArea(String group, String topic, int queueId, long offset) {
        //     if (!defaultMessageStore.getMessageStoreConfig().isColdDataFlowControlEnable()) {
        //         return false;
        //     }
        //     try {
        //         ConsumeQueue consumeQueue = (ConsumeQueue) defaultMessageStore.findConsumeQueue(topic, queueId);
        //         if (null == consumeQueue) {
        //             return false;
        //         }
        //         SelectMappedBufferResult bufferConsumeQueue = consumeQueue.getIndexBuffer(offset);
        //         if (null == bufferConsumeQueue || null == bufferConsumeQueue.getByteBuffer()) {
        //             return false;
        //         }
        //         long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();
        //         return defaultMessageStore.checkInColdAreaByCommitOffset(offsetPy, getMaxOffset());
        //     } catch (Exception e) {
        //         log.error("isMsgInColdArea group: {}, topic: {}, queueId: {}, offset: {}",
        //                 group, topic, queueId, offset, e);
        //     }
        //     return false;
        // }
    }


    public static class GroupCommitRequest {

        private final long nextOffset;

        private final CompletableFuture<PutMessageStatus> flushOKFuture = new CompletableFuture<>();

        private volatile int ackNums = 1;

        private final long deadLine;


        public GroupCommitRequest(long nextOffset, long timeoutMillis) {
            this.nextOffset = nextOffset;
            this.deadLine = System.nanoTime() + (timeoutMillis * 1_000_000);
        }


        public GroupCommitRequest(long nextOffset, long timeoutMillis, int ackNums) {
            this(nextOffset, timeoutMillis);
            this.ackNums = ackNums;
        }

        public long getNextOffset() {
            return nextOffset;
        }

        public int getAckNums() {
            return ackNums;
        }

        public long getDeadLine() {
            return deadLine;
        }


        public void wakeupCustomer(final PutMessageStatus status) {
            this.flushOKFuture.complete(status);
        }

        public CompletableFuture<PutMessageStatus> future() {
            return flushOKFuture;
        }
    }


    public void scanFileAndSetReadMode(int mode) {
        if (MixAll.isWindows()) {
            log.info("windows os stop scanFileAndSetReadMode");
            return;
        }
        try {
            log.info("scanFileAndSetReadMode mode: {}", mode);
            mappedFileQueue.getMappedFiles().forEach(mappedFile -> {
                setFileReadMode(mappedFile, mode);
            });
        } catch (Exception e) {
            log.error("scanFileAndSetReadMode exception", e);
        }
    }


    private int setFileReadMode(MappedFile mappedFile, int mode) {
        if (null == mappedFile) {
            log.error("setFileReadMode mappedFile is null");
            return -1;
        }
        final long address = ((DirectBuffer) mappedFile.getMappedByteBuffer()).address();
        int madvise = LibC.INSTANCE.madvise(new Pointer(address), new NativeLong(mappedFile.getFileSize()), mode);
        if (madvise != 0) {
            log.error("setFileReadMode error fileName: {}, madvise: {}, mode:{}",
                    mappedFile.getFileName(), madvise, mode);
        }
        return madvise;
    }

    public ColdDataCheckService getColdDataCheckService() {
        return coldDataCheckService;
    }


    @Override
    public void cleanSwappedMap(long forceCleanSwapIntervalMs) {

    }

    public long getConfirmOffset() {
        return getMaxOffset();
    }
    //根据消息在CommitLog文件中的偏移量和大小，得到消息内容的方法
    public SelectMappedBufferResult getData(final long offset) {
        //第二个参数意味着如果没找到对应的内存映射文件时是否返回第一个内存映射文件
        return this.getData(offset, offset == 0);
    }

    public SelectMappedBufferResult getData(final long offset, final boolean returnFirstOnNotFound) {
        int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
        MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, returnFirstOnNotFound);
        if (mappedFile != null) {
            int pos = (int) (offset % mappedFileSize);
            SelectMappedBufferResult result = mappedFile.selectMappedBuffer(pos);
            return result;
        }
        return null;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到消息重放请求的方法，方法的第一个参数byteBuffer就是封装了CommitLog文件中要重放的消息的缓冲区对象，外层方法会循环执行当前方法
     * 在当前方法中，byteBuffer也会不断读取自己内部的数据，直到读取完毕，没有数据可读取了
     */
    public DispatchRequest checkMessageAndReturnSize(java.nio.ByteBuffer byteBuffer, final boolean checkCRC,
                                                     final boolean checkDupInfo, final boolean readBody) {
        try {
            //读取缓冲区中首个消息的总长度，注意，这里读取了消息之后，指针就会往后便宜
            //所以只要外层方法还在循环执行，这里的byteBuffer就会不断读取自己内部数据
            int totalSize = byteBuffer.getInt();
            //得到消息的魔术
            int magicCode = byteBuffer.getInt();
            //根据魔数判断消息格式的合法性
            switch (magicCode) {
                //判断消息版本，如果是以下两个版本则不用处理，因为mq本身就支持这两个版本
                case MessageDecoder.MESSAGE_MAGIC_CODE:
                case MessageDecoder.MESSAGE_MAGIC_CODE_V2:
                    break;
                //如果是空魔术，则认为达到文件末尾，返回成功状态的并且为空的重放请求
                case BLANK_MAGIC_CODE:
                    return new DispatchRequest(0, true /* success */);
                //执行到这里意味着是非法的魔术，直接返回错误状态的重放请求
                default:
                    log.warn("found a illegal magic code 0x" + Integer.toHexString(magicCode));
                    return new DispatchRequest(-1, false /* success */);
            }
            //根据魔术得到消息版本
            MessageVersion messageVersion = MessageVersion.valueOfMagicCode(magicCode);
            //定义一个临时存储消息内容的字节数组
            byte[] bytesContent = new byte[totalSize];
            //得到消息的CRC校验码
            int bodyCRC = byteBuffer.getInt();
            //得到消息要存储的目标队列
            int queueId = byteBuffer.getInt();
            //得到消息标志位
            int flag = byteBuffer.getInt();
            //得到消息在消费队列中的偏移量
            long queueOffset = byteBuffer.getLong();
            //得到消息在CommitLog文件中的全局偏移量
            long physicOffset = byteBuffer.getLong();
            //得到消息的系统标志
            int sysFlag = byteBuffer.getInt();
            //得到消息的生产时间
            long bornTimeStamp = byteBuffer.getLong();
            //定义一个缓冲区，用来存储消息的生产服务器网络地址
            ByteBuffer byteBuffer1;
            //根据系统标志判断服务器地址的格式是否为IPv6地址
            if ((sysFlag & MessageSysFlag.BORNHOST_V6_FLAG) == 0) {
                //如果是IPv4地址，则直接从原始ByteBuffer中读取8字节，4字节IP + 4字节端口
                byteBuffer1 = byteBuffer.get(bytesContent, 0, 4 + 4);
            } else {
                //如果是IPv6地址，则直接从原始ByteBuffer中读取20字节，16字节IP + 4字节
                byteBuffer1 = byteBuffer.get(bytesContent, 0, 16 + 4);
            }
            //得到消息的存储时间
            long storeTimestamp = byteBuffer.getLong();
            //定义一个缓冲区，用来存储消息的存储服务器网络地址
            ByteBuffer byteBuffer2;
            //根据系统标志判断服务器地址的格式
            if ((sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
                byteBuffer2 = byteBuffer.get(bytesContent, 0, 4 + 4);
            } else {
                byteBuffer2 = byteBuffer.get(bytesContent, 0, 16 + 4);
            }
            //得到消息的重试次数
            int reconsumeTimes = byteBuffer.getInt();
            //得到事务消息的prepared偏移量
            long preparedTransactionOffset = byteBuffer.getLong();
            //得到消息体的长度
            int bodyLen = byteBuffer.getInt();
            if (bodyLen > 0) {
                //如果消息长度大于0，则判断本次操作是否需要读取消息体
                //因为在第十五版本代码中，当前方法只会在消息重放时被调用，而消息重放操作是不需要把消息体重放到消费队列中
                //所以这里并不会读取消息体内容
                if (readBody) {
                    //如果需要读取消息体内容，就在这里从缓冲区中读取
                    byteBuffer.get(bytesContent, 0, bodyLen);
                    //判断是否需要惊醒CRC校验，在第十五版本代码中也不需要进行校验
                    if (checkCRC) {
                        //如果需要校验，那就判断一下是否需要强制验证属性CRC
                        if (!this.defaultMessageStore.getMessageStoreConfig().isForceVerifyPropCRC()) {
                            //如果不需要则只验证消息体的CRC即可，不需要验证消息属性的CRC
                            //计算消息体的CRC32值
                            int crc = UtilAll.crc32(bytesContent, 0, bodyLen);
                            //比较计算得到的CRC值与存储的CRC值是否一致
                            if (crc != bodyCRC) {
                                //CRC校验失败，记录告警日志
                                log.warn("CRC check failed. bodyCRC={}, currentCRC={}", crc, bodyCRC);
                                //返回消息分发失败请求
                                return new DispatchRequest(-1, false/* success */);
                            }
                        }
                    }
                } else {
                    //如果不需要读取消息体，则跳过消息体字节
                    byteBuffer.position(byteBuffer.position() + bodyLen);
                }
            }
            //根据消息版本从缓冲区中得到消息主题的字节长度
            int topicLen = messageVersion.getTopicLength(byteBuffer);
            //从缓冲区中得到消息的字节数组
            byteBuffer.get(bytesContent, 0, topicLen);
            //得到消息主题信息
            String topic = new String(bytesContent, 0, topicLen, MessageDecoder.CHARSET_UTF8);
            //定义一个变量记录消息tag过滤标签的哈希码
            long tagsCode = 0;
            //初始化消息键
            String keys = "";
            //初始化消息唯一标识符
            String uniqKey = null;
            //读取消息属性长度
            short propertiesLength = byteBuffer.getShort();
            //初始化消息属性map，用于存储消息的所有属性键值对
            Map<String, String> propertiesMap = null;
            //判断消息是否存在属性数据
            if (propertiesLength > 0) {
                //读取消息属性内容到bytesContent数组
                byteBuffer.get(bytesContent, 0, propertiesLength);
                //得到消息属性的字符串内容
                String properties = new String(bytesContent, 0, propertiesLength, MessageDecoder.CHARSET_UTF8);
                //将消息属性字符串解析为map
                propertiesMap = MessageDecoder.string2messageProperties(properties);
                //得到消息的索引键
                keys = propertiesMap.get(MessageConst.PROPERTY_KEYS);
                //得到消息的唯一键
                uniqKey = propertiesMap.get(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
                //下面这段消息重复检测代码也不会执行到，源码中也不会用到，大家可以直接忽略
                if (checkDupInfo) {
                    String dupInfo = propertiesMap.get(MessageConst.DUP_INFO);
                    if (null == dupInfo || dupInfo.split("_").length != 2) {
                        log.warn("DupInfo in properties check failed. dupInfo={}", dupInfo);
                        return new DispatchRequest(-1, false);
                    }
                }
                //从消息属性中获取消息过滤标签
                String tags = propertiesMap.get(MessageConst.PROPERTY_TAGS);
                if (tags != null && tags.length() > 0) {
                    //标签不为空则计算哈希值
                    tagsCode = MessageExtBrokerInner.tagsString2tagsCode(MessageExt.parseTopicFilterType(sysFlag), tags);
                }
                {
                    //下面这个代码块执行的是和延迟消息相关的操作，这里就是从消息属性中得到消息的延迟级别
                    String t = propertiesMap.get(MessageConst.PROPERTY_DELAY_TIME_LEVEL);
                    //判断当前消息是否属于调度主题的消息，并且是否真的设置了延迟级别
                    //在mq中，延迟消息会被存储在RMQ_SYS_SCHEDULE_TOPIC程序内部主题中，并且延迟级别会保存在属性中
                    if (TopicValidator.RMQ_SYS_SCHEDULE_TOPIC.equals(topic) && t != null) {
                        //如果是延迟级别消息，那就在这里解析消息的延迟级别
                        int delayLevel = Integer.parseInt(t);
                        //判断消息的延迟级别是否超过系统允许的最大延迟级别，如果超过，则使用系统允许的最大延迟级别
                        if (delayLevel > this.defaultMessageStore.getMaxDelayLevel()) {
                            delayLevel = this.defaultMessageStore.getMaxDelayLevel();
                        }
                        //如果延迟级别大于0，则表示需要进行定时投递，也就是定时重放
                        //只有重放到消费队列中才可以被消费
                        if (delayLevel > 0) {
                            //计算消息实际投递时间，以实际投递时间作为消息tag哈希值
                            tagsCode = this.defaultMessageStore.computeDeliverTimestamp(delayLevel, storeTimestamp);
                        }
                    }
                }
            }
            //以下就是判断消息是否需要属性CRC校验的操作了，在第十五版本代码中这里仍然不需要执行校验操作
            if (checkCRC) {
                //判断是否需要强制对消息属性执行CRC校验
                if (this.defaultMessageStore.getMessageStoreConfig().isForceVerifyPropCRC()) {
                    int expectedCRC = -1;
                    //如果属性map不为空，则获取其CRC值
                    if (propertiesMap != null) {
                        //如果需要则在这里得到属性CRC32的值
                        String crc32Str = propertiesMap.get(MessageConst.PROPERTY_CRC32);
                        //如果CRC32字符串存在，将其转换为整数
                        if (crc32Str != null) {
                            expectedCRC = 0;
                            //从个位开始，依次解析每一位数字
                            //使用字符串从后向前遍历的方式进行解析
                            for (int i = crc32Str.length() - 1; i >= 0; i--) {
                                //将字符转换为对应的数字
                                int num = crc32Str.charAt(i) - '0';
                                //已有数值乘以10，相当于左移一位
                                expectedCRC *= 10;
                                //加上当前位的数字
                                expectedCRC += num;
                            }
                        }
                    }
                    if (expectedCRC > 0) {
                        //属性CRC解析成功则执行对比操作
                        ByteBuffer tmpBuffer = byteBuffer.duplicate();
                        //定位到消息的起始位置
                        tmpBuffer.position(tmpBuffer.position() - totalSize);
                        //设置缓冲器的结束位置
                        //结束位置为消息起始位置加上总大小，再减去CRC32保留长度
                        tmpBuffer.limit(tmpBuffer.position() + totalSize - CommitLog.CRC32_RESERVED_LEN);
                        //计算消息实际的CRC32值
                        int crc = UtilAll.crc32(tmpBuffer);
                        //比较CRC值与期望的CRC值
                        if (crc != expectedCRC) {
                            //校验失败则记录警告日志
                            log.warn(
                                    "CommitLog#checkAndDispatchMessage: failed to check message CRC, expected "
                                            + "CRC={}, actual CRC={}", bodyCRC, crc);
                            //返回失败状态的分发请求
                            return new DispatchRequest(-1, false/* success */);
                        }
                    } else {
                        //在属性中未找到CRC值，则记录警告日志
                        log.warn(
                                "CommitLog#checkAndDispatchMessage: failed to check message CRC, not found CRC in properties");
                        //返回失败状态的分发请求
                        return new DispatchRequest(-1, false/* success */);
                    }
                }
            }
            //计算消息的实际长度
            int readLength = MessageExtEncoder.calMsgLength(messageVersion, sysFlag, bodyLen, topicLen, propertiesLength);
            //判断最后解析出来的消息的实际长度和一开始读取到的是否一致
            if (totalSize != readLength) {
                //不一致则记录错误日志
                log.error(
                        "[BUG]read total count not equals msg total size. totalSize={}, readTotalCount={}, bodyLen={}, topicLen={}, propertiesLength={}",
                        totalSize, readLength, bodyLen, topicLen, propertiesLength);

                //返回失败状态的分发请求
                return new DispatchRequest(totalSize, false/* success */);
            }
            //执行到这里意味着本次消息分发成功了，创建消息分发请求对象即可
            DispatchRequest dispatchRequest = new DispatchRequest(
                    topic,
                    queueId,
                    physicOffset,
                    totalSize,
                    tagsCode,
                    storeTimestamp,
                    queueOffset,
                    keys,
                    uniqKey,
                    sysFlag,
                    preparedTransactionOffset,
                    propertiesMap
            );
            //返回成功创建的消息分发请求对象
            return dispatchRequest;

        } catch (Exception e) {
        }
        return new DispatchRequest(-1, false /* success */);

    }

}
